package skewreduce.framework.physical;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.LinkedBlockingQueue;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.JobStatus;
import org.apache.hadoop.mapreduce.Counter;
import org.apache.hadoop.mapreduce.CounterGroup;
import org.apache.hadoop.mapreduce.Counters;
import org.apache.hadoop.mapreduce.JobID;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import skewreduce.framework.SchedulerEvent;


/**
 * physical instantiation of a logical operator.
 *
 * success/failure: job end notification
 * blokcing task:   nothing to do
 * monitored task:  periodically check counters -- make local decision.
 *   monitored task may generate reoptimize event.
 */
public abstract class MonitoredPhysicalOp
extends PhysicalOp {
	private static Logger LOG = LoggerFactory.getLogger("PLAN");

	private long jobStatusPeriod;
	private long jobProgressPeriod;
	private long jobGracePeriod;
    private boolean mapOnly;

    protected boolean isRunning;
    private int wakeupSlack;

    private static WakeupMonitor wakeupMonitor;

    public static final WakeupMonitor getWakeupMonitor() {
        if ( wakeupMonitor == null ) {
            synchronized ( MonitoredPhysicalOp.class ) {
                if ( wakeupMonitor == null ) {
                    wakeupMonitor = new WakeupMonitor();
                }
            }
        }
        return wakeupMonitor;
    }

    public static class WakeupMonitor extends Thread {
        JobClient jobClient;
        LinkedBlockingQueue<MonitoredPhysicalOp> queue = new LinkedBlockingQueue<MonitoredPhysicalOp>();
        Set<MonitoredPhysicalOp> waiting = new HashSet<MonitoredPhysicalOp>();
        HashMap<JobID,JobStatus>   joinTable = new HashMap<JobID,JobStatus>();

        private WakeupMonitor() {
            super("skewreduce.wakeup.monitor");
            setDaemon(true);
        }

        public synchronized void init(Configuration conf) throws IOException {
            if ( jobClient == null ) {
                jobClient = new JobClient(new JobConf(conf));
                start();
                LOG.info("wakeup monitor has been initialized");
            }
        }

        public void register(MonitoredPhysicalOp op)
        throws InterruptedException {
            if ( op.getJobID() == null )
                throw new IllegalArgumentException();
            queue.put(op);
        }

        public void run() {
            LOG.info("wakeup monitor starts monitoring...");

            while ( true ) {
                queue.drainTo(waiting);
                if ( ! waiting.isEmpty() ) {
                    try {
                        // build a hash table
                        joinTable.clear();

                        for ( JobStatus status : jobClient.jobsToComplete() ) {
                            joinTable.put(status.getJobID(),status);
                        }

                        // join on status
                        Iterator<MonitoredPhysicalOp> i = waiting.iterator();
                        while ( i.hasNext() ) {
                            MonitoredPhysicalOp op = i.next();
                            JobID jobid = op.getJobID();
                            JobStatus status = joinTable.get(jobid);
                            if ( status == null ) {
                                // it's either not updated or not shown?
                                if ( ++op.wakeupSlack < 3 ) {
                                    LOG.info("Force waking up "+jobid);
                                    op.running(); // assumed something happened
                                    i.remove();
                                }
                            } else {
                                // good. check progress.
                                // if worker, wakeup when map progress start
                                // if merger, wakeup when reduce progress start

                                boolean wakeup = false;

                                wakeup = status.mapProgress() > 0.0f;
                                /*
                                if ( op instanceof PProcessOp ) {
                                    wakeup = status.mapProgress() > 0.0f;
                                } else {
                                    wakeup = status.reduceProgress() > 0.0f;
                                }
                                */

                                if ( wakeup ) {
                                    if ( LOG.isInfoEnabled() )
                                        LOG.info("waking up "+jobid);

                                    op.running();
                                    i.remove();
                                }
                            }
                        }
                        // alright. what's not waiting but prep?
                    } catch ( IOException ex ) {
                        LOG.info("wakeup mointor failed to contact job tracker",ex);
                    }
                }

                try {
                    Thread.sleep(5000); // run in every 5 seconds
                } catch ( InterruptedException ignore ) {}
            }
        }
    }

    public synchronized void running() {
        isRunning = true;
        notifyAll();
    }

    private void waitUntilRunning() throws InterruptedException {
        wakeupMonitor.register(this);

        while ( ! isRunning ) {
            synchronized (this) {
                wait();
            }
        }
    }
    
    @Override
	public void setConf(Configuration conf) {
		super.setConf(conf);
        jobStatusPeriod = getConf().getLong("skewreduce.monitoring.job.status",5000);
        jobProgressPeriod = getConf().getLong("skewreduce.monitoring.job.progress",1 * 60 * 1000);
        jobGracePeriod = getConf().getLong("skewreduce.monitoring.job.graceperiod",2 * 60 * 1000);
        mapOnly = getConf().getBoolean("skewreduce.merge.mapOnly",false);
	}

    protected void updateStatistics(CounterGroup group,long lidx,boolean done,boolean reactSkew) {
        // by default, just print out counters
        if ( LOG.isDebugEnabled() ) {
            for ( Counter cnt : group ) {
                LOG.debug(String.format("%s = %d",cnt.getName(),cnt.getValue()));
            }
        }

        if ( done ) {
            setStatistics(group);
            if ( mapOnly || this instanceof PProcessOp ) {
                long b = group.findCounter("MAP_BEGIN").getValue();
                long e = group.findCounter("MAP_END").getValue();

                LOG.info(String.format("EAGG-MR\t%s\t%s\t%d\t%d",getID(),job.getJobID(),b,e));
            } else {
                long b = group.findCounter("REDUCE_BEGIN").getValue();
                long e = group.findCounter("REDUCE_END").getValue();

                LOG.info(String.format("EAGG-MR\t%s\t%s\t%d\t%d",getID(),job.getJobID(),b,e));
            }

            /*
            if ( this instanceof PMergeOp ) {
                long b = group.findCounter("REDUCE_BEGIN").getValue();
                long e = group.findCounter("REDUCE_END").getValue();

                LOG.info(String.format("EAGG-MR\t%s\t%s\t%d\t%d",getID(),job.getJobID(),b,e));
            } else if ( this instanceof PProcessOp ) {
                long b = group.findCounter("MAP_BEGIN").getValue();
                long e = group.findCounter("MAP_END").getValue();

                LOG.info(String.format("EAGG-MR\t%s\t%s\t%d\t%d",getID(),job.getJobID(),b,e));
            }
            */
        }
    }

    private SchedulerEvent monitor() throws InterruptedException {
        long currentTime = System.currentTimeMillis();
        long nextCheckTime    = currentTime + jobStatusPeriod;
        long nextCollectTime  = currentTime + jobProgressPeriod;
        long endOfGracePeriod = currentTime + jobGracePeriod;

        boolean done = false;
        boolean doMonitor = false;
        boolean okayToFireSlowRunning = false;

        while ( true ) {
            currentTime = System.currentTimeMillis();
            try {
                if ( nextCheckTime <= currentTime ) {
                    nextCheckTime += jobStatusPeriod;
                    done = isComplete();
                }
                Counters counters = job.getCounters();
                CounterGroup cntGroup = counters.getGroup("skewreduce");
                Counter loopIdx = cntGroup.findCounter("IDX");
                long lidx = loopIdx.getValue();

                // check whether index variable has none-zero values
                if ( ! doMonitor && lidx > 0 ) {
                    doMonitor = true;
                    endOfGracePeriod = System.currentTimeMillis() + jobGracePeriod;
                    LOG.info("switch to monitor mode: "+lidx);
                }

                if ( ! okayToFireSlowRunning ) {
                    okayToFireSlowRunning = ( endOfGracePeriod < currentTime );
                    if ( okayToFireSlowRunning )
                        LOG.info("switch to monitor-and-react mode: "+lidx);
                }

                if ( doMonitor ) {
                    // process counters
                    // lidx, cntGroup, done, okayToFireSlowRunning
                    updateStatistics(cntGroup,lidx,done,okayToFireSlowRunning);
                }

                if ( okayToFireSlowRunning && ! done ) {
                    if ( LOG.isDebugEnabled() ) {
                        LOG.debug("testing slowness");
                    }
                    // FIXME: test slowness of this task
                }

                if ( done ) break;  // we are done
            } catch ( IOException ex ) {
                if ( Thread.interrupted() )
                    throw new InterruptedException();

                LOG.warn( job.toString(), ex );
            }

            long sleepTime = nextCheckTime - System.currentTimeMillis();
            if ( sleepTime > 0 ) {
                Thread.sleep( sleepTime );
            } else {
                LOG.warn(String.format("Missed period by %d ms",sleepTime*-1));
            }
        }

        return ( isSuccessful() )
                ? new SchedulerEvent.Completion(this) 
                : new SchedulerEvent.Failed(this);
    }

    public SchedulerEvent call() throws Exception {
        SchedulerEvent result;
        
        if ( LOG.isInfoEnabled() ) {
        	LOG.info("EXECUTE {}",getID());
        }

        try {
        	setup();
        	
        	scheduledAt = System.currentTimeMillis();
        	
            job.submit();

            //plan.registerRunningPhysicalOp(this);

            setStatus(Status.RUNNING);

            /*
            JobID jobid = job.getJobID();

            if ( LOG.isInfoEnabled() ) {
                LOG.info(String.format("EAGG-MR\t%s\t%s",getID(),jobid));
            }
            */

            waitUntilRunning();

            result = monitor();
        } catch ( InterruptedException x ) {
            if ( Thread.interrupted() ) {
                LOG.info("Killing MapReduce task : "+job.getJobID());
                try { job.killJob(); } catch ( IOException ignore ) {}
                return new SchedulerEvent.Killed(this);
            } else {
                return new SchedulerEvent.Error(this,x);
            }
        } catch ( Exception x ) {
            LOG.error("Failed due to exception",x);
            return new SchedulerEvent.Error(this,x);
        }

        //return result;
        return new SchedulerEvent.Noop(this);
    }
}
